/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.wx.mq.mqservice;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import com.alibaba.fastjson.JSONObject;
import java.util.HashMap;
import org.apache.log4j.Logger;
import cn.ac.iie.di.wx.mq.common.RuntimeEnv;
import static cn.ac.iie.di.wx.mq.common.RuntimeEnv.fhMaxNum;
import cn.ac.iie.di.wx.mq.httpservice.HttpSender;
import cn.ac.iie.di.wx.mq.utils.TimeStampUtil;
import cn.ac.iie.di.wx.mq.utils.ToJsonUtil;
import cn.ac.iie.di.wx.mq.vo.RedisVo;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 *
 * @author Administrator
 */
public class ReceiveHandler extends REAbstractReceiveMessageHandler<REMessage> {

    static Logger logger = Logger.getLogger(ReceiveHandler.class.getName());
    private TimeStampUtil stampUtil;
    private LinkedBlockingQueue<HashMap<String, String>> redisQueue;

    @Override
    public boolean handle(REMessage message) {
        try {

            int g_id_index = 0;
            int m_mm_url_index = 0;
            //添加4个字段的入redis
            int u_ch_id_index = 0;
            int m_ch_id_index = 0;
            int m_chat_room_index = 0;
            int table_index = 0;
            for (int i = 0; i < message.colNames.length; i++) {
//            logger.info("column:" + message.colNames[i] + "\t");
                if (message.colNames[i].equals("g_id")) {
                    g_id_index = i;
                }
                if (message.colNames[i].equals("m_mm_url")) {
                    m_mm_url_index = i;
                }
                if (message.colNames[i].equals("u_ch_id")) {
                    u_ch_id_index = i;
                }
                if (message.colNames[i].equals("m_ch_id")) {
                    m_ch_id_index = i;
                }
                if (message.colNames[i].equals("m_chat_room")) {
                    m_chat_room_index = i;
                }
                if (message.colNames[i].equals("table")) {
                    table_index = i;
                }
            }
//            logger.info("get data size:" + message.data.size());
            //存redis的hashbao
            HashMap<String, String> g_idAndopidMap = new HashMap<String, String>();
            for (Object[] col : message.data) {
                try {
                    if (col == null) {
                        logger.info("(null)\t");
                        continue;
                    }
                    //获取g_id，m_mm_url，u_ch_id,m_ch_id,m_chat_room,table,下载多媒体文件暂时不定字段类型
                    String g_id = col[g_id_index].toString();
                    String m_mm_url = col[m_mm_url_index].toString();
                    String u_ch_id = col[u_ch_id_index].toString();
                    String m_ch_id = col[m_ch_id_index].toString();
                    String m_chat_room = col[m_chat_room_index].toString();
                    String table = col[table_index].toString();

                    long xuleihao = stampUtil.nextId();//不知道英文怎么说
                    String opid = "133" + "440300" + String.valueOf(xuleihao);//最后位数最多为10位
                    long sendtime = System.currentTimeMillis() / 1000;
                    String g_id_time = g_id + "_" + sendtime;
                    String redisValue = "{\"g_id\":\"" + g_id + "\",\"sendtime\":" + sendtime + ",\"table\":\"" + table + "\",\"m_ch_id\":\"" + m_ch_id + "\",\"u_ch_id\":\"" + u_ch_id + "\",\"m_chat_room\":\"" + m_chat_room + "\"}";
                    g_idAndopidMap.put(opid, redisValue);
                    logger.info("put into map ,opid is " + opid + ",and g_id_time is " + g_id_time + ",json String is "
                            + redisValue);
                    if (!g_id.equals("") && !m_mm_url.equals("") && !u_ch_id.equals("") && !m_ch_id.equals("") && !m_chat_room.equals("") && !table.equals("")) {
                        RuntimeEnv.CSCount.getAndIncrement();
                        //发送给烽火，数字从我这边生成，当前，redis存放key=opid,value=g_id,opid由原子变量控制的一个毫秒级的时间戳
                        String json = ToJsonUtil.jsonUtil(opid, "440300", "440300", "WXB", opid, "admin", opid, "url", m_mm_url);
                        logger.info("groupMedia request :" + json);
                        String result = "";
                        while (true) {
                            if (RuntimeEnv.FLLIMIT.get() <= Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.fhMaxNum))) {
                                RuntimeEnv.FLLIMIT.getAndIncrement();
                                long start = System.currentTimeMillis();
                                result = HttpSender.sendStringByUrl(json, (String) RuntimeEnv.getParam(RuntimeEnv.groupMediaQuery));
                                logger.info("get response use time = " + (System.currentTimeMillis()- start) + " ms");
                                try {
                                    Thread.sleep(1000);
                                } catch (InterruptedException ex) {
                                    logger.error(ex.getMessage(), ex);
                                }
                                RuntimeEnv.FLLIMIT.getAndDecrement();
                                break;
                            }
                            Thread.sleep(1);
                        }
                        logger.info("request is : " + json + ",send to " + (String) RuntimeEnv.getParam(RuntimeEnv.groupMediaQuery)
                                + ",response is " + result + ",g_id_time is " + g_id_time + ",json String is "
                                + redisValue);
                        RedisVo redisVo = null;
                        if (result.equals("")) {
                            //http异常
                            RuntimeEnv.FLCount.getAndIncrement();
                            logger.info("failed groupMedia is null,put into retry MQueue");
                            redisVo = new RedisVo(opid, redisValue, json);
                        } else {
                            JSONObject jSONObject = JSONObject.parseObject(result);
                            if (jSONObject.getString("status").equals("200")) {
                                //操作成功
                                logger.info("groupMedia response is:" + result);
                                RuntimeEnv.GMCount.getAndIncrement();
                            } else {
                                RuntimeEnv.FLCount.getAndIncrement();
                                logger.info("failed groupMedia is not 200,put into retry MQueue");
                                redisVo = new RedisVo(opid, redisValue, json);
                            }
                        }
                        if (redisVo != null) {
                            while (true) {
                                try {
                                    if (RuntimeEnv.retryQueue.offer(redisVo)) {
                                        break;
                                    }
                                    Thread.sleep(200);
                                } catch (Exception e) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                        }
                    } else {
                        logger.warn("g_id :" + g_id + ",m_mm_url : " + m_mm_url + ",u_ch_id :" + u_ch_id + ",m_ch_id : " + m_ch_id + ",m_chat_room : :" + m_chat_room + ",table :" + table);
                    }
                } catch (Exception ex) {
                    logger.error(ex.getMessage(), ex);
                }

            }
            //尝试入队列直至成功
            while (true) {
                try {
                    if (g_idAndopidMap == null || redisQueue.offer(g_idAndopidMap)) {
                        break;
                    }
                    Thread.sleep(200);
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return true;
    }

    public ReceiveHandler(TimeStampUtil stampUtil, LinkedBlockingQueue<HashMap<String, String>> redisQueue) {
        this.stampUtil = stampUtil;
        this.redisQueue = redisQueue;
    }

    public TimeStampUtil getStampUtil() {
        return stampUtil;
    }

    public void setStampUtil(TimeStampUtil stampUtil) {
        this.stampUtil = stampUtil;
    }

    public LinkedBlockingQueue<HashMap<String, String>> getRedisQueue() {
        return redisQueue;
    }

    public void setRedisQueue(LinkedBlockingQueue<HashMap<String, String>> redisQueue) {
        this.redisQueue = redisQueue;
    }

}
